package com.zyx.pulsardemo.basic.readers;

import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Reader;

import java.util.Date;

/**
 * @author Yaxi.Zhang
 * @since 2021/9/30 18:07
 * reference: https://github.com/david-streamlio/pulsar-in-action
 */
public class PulsarReaders {

	private static Producer<byte[]> producer;
    private static Reader<byte[]> reader;

    public static void main(String[] args) throws Exception {
		PulsarClient client = PulsarClient.builder()
				.serviceUrl("pulsar://bigdata022:6650,bigdata023:6650,bigdata024:6650")
				.build();

        producer = client.newProducer()
				// 不指定 tenant 及 namespace 时 默认使用 persistent://public/default/my-other-topic
                .topic("my-other-topic")
                .create();

        reader = client.newReader()
				// 不指定 tenant 及 namespace 时 默认使用 persistent://public/default/my-other-topic
                .topic("my-other-topic")
                .readerName("my-reader")
                .startMessageId(MessageId.earliest)
                .create();

        // startProducer();
        startReader();
    }

    private static void startProducer() throws Exception {
        int counter = 0;
        while (true) {
            String msg = "my-message-" + (++counter) + "---" + new Date();
            producer.newMessage()
                    .value(msg.getBytes())
                    .send();

            System.out.println("send-message..." + msg);

            Thread.sleep(1000);
        }
    }

    private static void startReader() throws PulsarClientException {
        MessageId lastRead = null;
        do {
            Message<byte[]> msg = reader.readNext();
            System.out.printf("Message read: %s \n", new String(msg.getData()));
            lastRead = msg.getMessageId();
        } while (!reader.hasReachedEndOfTopic());
    }

}
